package org.netty.framework.core.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * netty server 的业务线程池
 * 
 * @author chenfanglin
 * @date 2018年9月20日下午5:42:32
 */
public class NettyServerThreadPoolExecutor extends NettyServerThreadFactory {

	private static final String POOL_NAME = "netty-server";

	public NettyServerThreadPoolExecutor() {
		this(POOL_NAME);
	}

	public NettyServerThreadPoolExecutor(String poolName) {
		super(poolName);
		initializeExecutor();
	}

	private final Object poolSizeMonitor = new Object();

	/**
	 * 线程池最小的核心线程数
	 */
	private int corePoolSize = 1;

	/**
	 * 线程池最大的核心线程数
	 */
	private int maxPoolSize = Integer.MAX_VALUE;

	/**
	 * 空闲线程的存活时间(单位秒)
	 */
	private int keepAliveSeconds = 60;

	/**
	 * 队列长度
	 */
	private int queueCapacity = Integer.MAX_VALUE;

	/**
	 * jdk的线程池
	 */
	private ThreadPoolExecutor threadPoolExecutor;

	/**
	 * Reject策略预定义有四种： (1)ThreadPoolExecutor.AbortPolicy策略，是默认的策略,处理程序遭到拒绝将抛出运行时
	 * RejectedExecutionException。 (2)ThreadPoolExecutor.CallerRunsPolicy策略
	 * ,调用者的线程会执行该任务,如果执行器已关闭,则丢弃.
	 * (3)ThreadPoolExecutor.DiscardPolicy策略，不能执行的任务将被丢弃.
	 * (4)ThreadPoolExecutor.DiscardOldestPolicy策略，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）.
	 */
	private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();

	/**
	 * Set the ThreadPoolExecutor's core pool size. Default is 1.
	 * <p>
	 * <b>This setting can be modified at runtime, for example through JMX.</b>
	 */
	public void setCorePoolSize(int corePoolSize) {
		synchronized (this.poolSizeMonitor) {
			this.corePoolSize = corePoolSize;
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setCorePoolSize(corePoolSize);
			}
		}
	}

	/**
	 * Return the ThreadPoolExecutor's core pool size.
	 */
	public int getCorePoolSize() {
		synchronized (this.poolSizeMonitor) {
			return this.corePoolSize;
		}
	}

	/**
	 * Set the ThreadPoolExecutor's maximum pool size. Default is
	 * {@code Integer.MAX_VALUE}.
	 * <p>
	 * <b>This setting can be modified at runtime, for example through JMX.</b>
	 */
	public void setMaxPoolSize(int maxPoolSize) {
		synchronized (this.poolSizeMonitor) {
			this.maxPoolSize = maxPoolSize;
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
			}
		}
	}

	/**
	 * Return the ThreadPoolExecutor's maximum pool size.
	 */
	public int getMaxPoolSize() {
		synchronized (this.poolSizeMonitor) {
			return this.maxPoolSize;
		}
	}

	/**
	 * Set the ThreadPoolExecutor's keep-alive seconds. Default is 60.
	 * <p>
	 * <b>This setting can be modified at runtime, for example through JMX.</b>
	 */
	public void setKeepAliveSeconds(int keepAliveSeconds) {
		synchronized (this.poolSizeMonitor) {
			this.keepAliveSeconds = keepAliveSeconds;
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
			}
		}
	}

	/**
	 * Return the ThreadPoolExecutor's keep-alive seconds.
	 */
	public int getKeepAliveSeconds() {
		synchronized (this.poolSizeMonitor) {
			return this.keepAliveSeconds;
		}
	}

	/**
	 * Set the capacity for the ThreadPoolExecutor's BlockingQueue. Default is
	 * {@code Integer.MAX_VALUE}.
	 * <p>
	 * Any positive value will lead to a LinkedBlockingQueue instance; any other
	 * value will lead to a SynchronousQueue instance.
	 * 
	 * @see java.util.concurrent.LinkedBlockingQueue
	 * @see java.util.concurrent.SynchronousQueue
	 */
	public void setQueueCapacity(int queueCapacity) {
		this.queueCapacity = queueCapacity;
	}

	/**
	 * 初始化线程池
	 */
	private void initializeExecutor() {
		BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
		threadPoolExecutor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds,
				TimeUnit.SECONDS, queue, this, rejectedExecutionHandler);
	}

	/**
	 * 创建一个任务队列
	 * 
	 * @param queueCapacity
	 * @return
	 */
	private BlockingQueue<Runnable> createQueue(int queueCapacity) {
		if (queueCapacity > 0) {
			return new LinkedBlockingQueue<>(queueCapacity);
		} else {
			return new SynchronousQueue<>();
		}
	}

	/**
	 * Return the current pool size.
	 * 
	 * @see java.util.concurrent.ThreadPoolExecutor#getPoolSize()
	 */
	public int getPoolSize() {
		if (this.threadPoolExecutor == null) {
			// Not initialized yet: assume core pool size.
			return this.corePoolSize;
		}
		return this.threadPoolExecutor.getPoolSize();
	}

	/**
	 * 获取线程池
	 * 
	 * @return
	 * @throws IllegalStateException
	 */
	public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException {
		if (threadPoolExecutor == null) {
			throw new IllegalStateException("NettyServerThreadPoolExecutor not initialized");
		}
		return threadPoolExecutor;
	}

	public void execute(Runnable task) {
		Executor executor = getThreadPoolExecutor();
		try {
			executor.execute(task);
		} catch (RejectedExecutionException ex) {
			throw ex;
		}
	}

	public Future<?> submit(Runnable task) {
		ExecutorService executor = getThreadPoolExecutor();
		try {
			return executor.submit(task);
		} catch (RejectedExecutionException ex) {
			throw ex;
		}
	}

	public <T> Future<T> submit(Callable<T> task) {
		ExecutorService executor = getThreadPoolExecutor();
		try {
			return executor.submit(task);
		} catch (RejectedExecutionException ex) {
			throw ex;
		}
	}

}
